package com.canal.util;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;

import javax.sql.DataSource;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

public class CanalUtils {
  /**
   * 获取canal连接器
   */
  public static CanalConnector getSingleCanalConnector(String host, int port, String desc, String user, String password) {
    CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(host, port), desc, user, password);
    return connector;
  }

  /**
   * 获取集群canal连接器
   */
  public static CanalConnector getClusterCanalConnector(List<? extends SocketAddress> addresses, String desc, String user, String password) {
    CanalConnector connector = CanalConnectors.newClusterConnector(addresses, desc, user, password);
    return connector;
  }

  public static CanalConnector getClusterCanalConnectorV2(List<Map<String,String>> adrs, String desc, String user, String password) {
    List<InetSocketAddress> addresses = new ArrayList<>();
    adrs.forEach(en -> {
      String host = en.get("host");
      String port = en.get("port");
      addresses.add(new InetSocketAddress(host, Integer.parseInt(port)));
    });
    CanalConnector connector = CanalConnectors.newClusterConnector(addresses, desc, user, password);
    return connector;
  }

  /**
   * 获取由zookeeper组成的canal集群链接
   */
  public static CanalConnector getClusterCanalConnector(String zkHost, String desc, String user, String password) {
    CanalConnector connector = CanalConnectors.newClusterConnector(zkHost, desc, user, password);
    return connector;
  }

  public static CanalConnector getLocalCanalConnector() {
    //        return getCanalConnector(AddressUtils.getHostIp(), 11111, "example", "", "");
    return getSingleCanalConnector("127.0.0.1", 11111, "example", "", "");
  }

  /**
   * 获取MySQL连接
   */
  public static Connection getMySqlConnnection(String url, String userName, String password) {
    Connection connection = null;
    try {
      Class.forName("com.mysql.jdbc.Driver");
      connection = DriverManager.getConnection(url, userName, password);
    } catch (Exception e) {
      System.out.println(e.getLocalizedMessage());
    }
    return connection;
  }

  public static boolean execute(Connection conn, String sql) {
    try (PreparedStatement statement = conn.prepareStatement(sql)) {
      return statement.execute();
    } catch (Exception e) {
      System.out.println(e.getLocalizedMessage());
    }
    return false;
  }

  /**
   * 获取数据连接池
   */
  public static DataSource getDataSource(String url, String user, String psd) {
    DruidDataSource dataSource = new DruidDataSource();
    // 2、为数据源实例指定必须的属性
    dataSource.setUsername(user);
    dataSource.setPassword(psd);
    dataSource.setUrl(url);
    dataSource.setDriverClassName("com.mysql.jdbc.Driver");
    // 3、指定数据源的一些可选属性
    // 3.1、指定数据库连接池中出使化连接数的个数
    dataSource.setInitialSize(5);
    // 3.2、指定最大连接数：同意时刻可以同时向数据库申请的连接数
    dataSource.setMaxActive(20);
    // 3.3、指定最小连接数:在数据库连接池中保存的最少的空闲连接的数量
    dataSource.setMinIdle(2);
    // 3.4、等待数据库连接池分配连接的最长时间。单位为毫秒。超出时间将抛出异常
    dataSource.setMaxWait(1000 * 5);
    // 4、从数据源中获取数据库连接
    return dataSource;
  }

  /**
   * 数据连接池获取链接
   */
  public static Connection getConnFromPool(DataSource dataSource) throws Exception {
    return dataSource.getConnection();
  }
}